/*
 * Copyright (C) 2012  Romain DUBOIS
 * 
 * This file is part of NeMoS - Network and Modular Software
 * 
 * NeMoS is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 * 
 * NeMoS is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with NeMoS.  If not, see <http://www.gnu.org/licenses/>.
 */
package nemos.remote.rsa.message.impl;

import static nemos.remote.rsa.message.data.EFrameworkAction.JOIN;
import static nemos.remote.rsa.message.data.EFrameworkAction.LEAVE;

import java.beans.EventHandler;
import java.io.IOException;
import java.rmi.RemoteException;
import java.util.Collections;
import java.util.Hashtable;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import nemos.remote.rsa.internal.RsaMessage;
import nemos.remote.rsa.message.IConnection;
import nemos.remote.rsa.message.data.EndpointMessage;
import nemos.remote.rsa.message.data.FrameworkMessage;
import nemos.remote.rsa.message.data.InitMessage;
import nemos.remote.rsa.message.data.Message;
import nemos.remote.rsa.message.transport.ITransport;
import nemos.remote.util.RemoteUtil;
import nemos.util.log.NemosLogger;

import org.osgi.framework.BundleContext;
import org.osgi.framework.ServiceException;
import org.osgi.framework.ServiceFactory;
import org.osgi.framework.ServiceRegistration;
import org.osgi.service.remoteserviceadmin.EndpointDescription;
import org.osgi.service.remoteserviceadmin.EndpointListener;
import org.osgi.service.remoteserviceadmin.RemoteServiceAdmin;

/**
 * ConnectionImpl : The connection supervisor class.
 * 
 * <p>
 * This class handles a connection :
 * <ul>
 * <li>It listens to incoming messages and forward them to the
 * {@link MessageProcessor},</li>
 * <li>It listens to auto-export endpoints and notify remote server</li>
 * <li>It registers a {@link ServiceFactory} for transport-based
 * {@link RemoteServiceAdmin}</li>
 * </ul>
 * 
 * It is responsible of those components initialization and shutdown.
 * </p>
 * 
 * @author Romain DUBOIS
 */
public class ConnectionImpl implements IConnection, EndpointListener {

    /** The bundle context */
    private BundleContext mContext;

    /** The listening thread */
    private final Thread mListeningThread;
    /** The logger */
    private final NemosLogger mLogger;

    /** The service registration */
    private ServiceRegistration<?> mRegistration;
    /** The transport to use */
    private final ITransport mTransport;
    /** The message processor */
    private final MessageProcessor mProcessor;
    /** The RSA factory */
    private final TransportRsaFactory mRsaFactory;

    /** The active flag */
    private boolean mActive;

    /**
     * Constructor
     * 
     * @param pTransport
     *            The transport to use
     */
    public ConnectionImpl(final ITransport pTransport) {
        super();

        mTransport = pTransport;
        mLogger = new NemosLogger(ConnectionImpl.class);
        mRsaFactory = new TransportRsaFactory(mTransport);
        mProcessor = new MessageProcessor(mTransport);
        mListeningThread = new Thread(EventHandler.create(Runnable.class, this, "listen")); //$NON-NLS-1$
    }

    /**
     * {@inheritDoc}
     * 
     * @see org.osgi.service.remoteserviceadmin.EndpointListener#endpointAdded(org.osgi.service.remoteserviceadmin.EndpointDescription,
     *      java.lang.String)
     */
    @Override
    public void endpointAdded(final EndpointDescription pEndpoint,
            @SuppressWarnings("unused") final String pMatchedFilter) {
        sendEpMessage(pEndpoint, true);
    }

    /**
     * {@inheritDoc}
     * 
     * @see org.osgi.service.remoteserviceadmin.EndpointListener#endpointRemoved(org.osgi.service.remoteserviceadmin.EndpointDescription,
     *      java.lang.String)
     */
    @Override
    public void endpointRemoved(final EndpointDescription pEndpoint,
            @SuppressWarnings("unused") final String pMatchedFilter) {
        sendEpMessage(pEndpoint, false);
    }

    /**
     * Send an endpoint message.
     * 
     * @param pEndpoint
     *            The endpoint
     * @param pAdded
     *            The add flag
     */
    private void sendEpMessage(final EndpointDescription pEndpoint, final boolean pAdded) {
        final EndpointMessage lMsg = new EndpointMessage(pEndpoint, pAdded);
        try {
            mTransport.send(lMsg);
        } catch (final RemoteException e) {
            mLogger.error(RsaMessage.SENDING_ERROR, e, lMsg);
        }
    }

    /**
     * {@inheritDoc}
     * 
     * @see nemos.remote.rsa.message.IConnection#getConnectedFrameworks()
     */
    @Override
    public Set<String> getConnectedFrameworks() {
        return Collections.unmodifiableSet(mProcessor.getConnectedFrameworks());
    }

    /**
     * {@inheritDoc}
     * 
     * @see nemos.remote.rsa.message.IConnection#getId()
     */
    @Override
    public String getId() {
        return mTransport.getId();
    }

    /**
     * Listen to the transport layer.
     */
    public void listen() {
        mLogger.debug(RsaMessage.START_LISTENING);
        try {

            // Listen continuously, until transport or service shutdown
            while (!Thread.interrupted() && mTransport.isActive()) {
                try {
                    final Message lMessage = mTransport.receive(500, TimeUnit.MILLISECONDS);

                    // Message can be null if timeout is reached
                    if (lMessage != null) {
                        if (mProcessor.process(lMessage)) {
                            mTransport.close();
                        }
                    }
                } catch (final RemoteException e) {
                    mLogger.error(RsaMessage.LISTENING_EXCEPTION, e);
                }
            }

        } catch (final InterruptedException e) {
            // Thread interrupted : stop listening
        } finally {
            mLogger.debug(RsaMessage.STOP_LISTENING);
            stop();
        }
    }

    /**
     * Start the connection.
     * 
     * @param pContext
     *            The bundle context
     * @throws IOException
     *             If start failed
     */
    public synchronized void start(final BundleContext pContext) throws IOException {
        mContext = pContext;

        // First, prepare message processor and provide it the init message
        mProcessor.start(mContext);
        try {
            final Message lMessage = mTransport.receive(2, TimeUnit.MINUTES);
            if (lMessage instanceof InitMessage) {
                mProcessor.process(lMessage);
            } else {
                mProcessor.stop();
                throw new IOException(mLogger.getTranslator().getMessage(RsaMessage.INVALID_INIT,
                        lMessage.toString()));
            }
        } catch (final Exception e) {
            mProcessor.stop();
            throw new IOException(mLogger.getTranslator().getMessage(RsaMessage.INIT_FAILED), e);
        }

        // Start listening to incoming messages
        mListeningThread.start();

        // Notify framework support
        try {
            mTransport.send(new FrameworkMessage(RemoteUtil.getFrameworkUuid(), JOIN));
        } catch (final RemoteException e) {
            stop();
            throw new ServiceException(mLogger.getTranslator().getMessage(RsaMessage.START_FAILED),
                    e);
        }

        // Now that communication is ready, register services for external use
        // Connection at least (registered = active)
        mRsaFactory.start(pContext);
        final Hashtable<String, Object> lProps = new Hashtable<String, Object>();
        lProps.put(ENDPOINT_LISTENER_SCOPE, RemoteUtil.getLocalEndpointFilter());
        mRegistration = pContext.registerService(new String[] { IConnection.class.getName(),
                EndpointListener.class.getName() }, this, lProps);
        mLogger.setServiceReference(mRegistration.getReference());

        mActive = true;
    }

    /**
     * {@inheritDoc}
     * 
     * @see nemos.remote.rsa.message.IConnection#stop()
     */
    @Override
    public void stop() {

        boolean lWaitForListeningThread = false;
        synchronized (this) {
            if (mActive) {
                mActive = false;
                // First unregister in order to deactivate associated RSAs
                // Imported services will be unregistered too and above all,
                // exported
                // services will be unexported
                mRsaFactory.stop();
                if (mRegistration != null) {
                    // If start fails, it can call stop with no registration
                    mRegistration.unregister();
                    mRegistration = null;
                }

                // Then unregister framework until mtransport is closed
                if (mTransport.isActive()) {
                    try {
                        mTransport.send(new FrameworkMessage(RemoteUtil.getFrameworkUuid(), LEAVE));
                    } catch (final RemoteException e) {
                        mLogger.error(RsaMessage.STOP_FAILED, e);
                    }
                }

                // Next, stop listening incoming messages
                if (mListeningThread.isAlive() && mListeningThread != Thread.currentThread()) {
                    mListeningThread.interrupt();
                    lWaitForListeningThread = true;
                }
            }
        }

        if (lWaitForListeningThread) {
            try {
                mListeningThread.join();
            } catch (final InterruptedException e) {
                mLogger.warn(RsaMessage.INTERRUPTED_THREAD, e);
            }
        }

        // No incoming message can arrive now : listening thread has been
        // interrupted
        mProcessor.stop();

        // Last : close transport
        mTransport.close();
    }

    /**
     * Get the active.
     * 
     * @return The active
     */
    public boolean isActive() {
        return mActive;
    }
}
